package com.bawei.flink.tableapi;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class IntervalJoinDemo {

    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();

        TableEnvironment tEnv = TableEnvironment.create(settings);

        tEnv.executeSql("create table shipments(\n" +
                "id BIGINT,\n" +
                "name STRING,\n" +
                "sts BIGINT,\n" +
                "sts_ltz AS TO_TIMESTAMP_LTZ(sts,3),\n" +
                "WATERMARK FOR sts_ltz AS sts_ltz - INTERVAL '30' MINUTE,\n" +
                "PRIMARY KEY(id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456',\n" +
                "  'driver' = 'com.mysql.jdbc.Driver',\n" +
                "  'url' = 'jdbc:mysql://cdh1:3306/mydatabase',\n" +
                "  'table-name' = 'shipments')");

        tEnv.executeSql("create table orders(\n" +
                "order_id BIGINT,\n" +
                "order_name STRING,\n" +
                "shipments_id BIGINT,\n" +
                "ots BIGINT,\n" +
                "ots_ltz AS TO_TIMESTAMP_LTZ(ots,3),\n" +
                "WATERMARK FOR ots_ltz AS ots_ltz - INTERVAL '30' MINUTE,\n" +
                "PRIMARY KEY(order_id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456',\n" +
                "  'driver' = 'com.mysql.jdbc.Driver',\n" +
                "  'url' = 'jdbc:mysql://cdh1:3306/mydatabase',\n" +
                "  'table-name' = 'orders')");


        tEnv.sqlQuery("select o.order_id order_id, o.order_name order_name,s.name shipments_name from orders o,shipments s where o.shipments_id = s.id and o.ots_ltz between s.sts_ltz - INTERVAL '4' HOUR AND s.sts_ltz + INTERVAL '1' HOUR").execute().print();


    }

}
